package com.xxd.dfs.backupnode;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.xxd.dfs.namenode.rpc.model.FetchEditLogResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author: XiaoDong.Xie
 * @create: 2020-08-25 10:06
 * @description: 这是一个实时去namenode 拉取的组件
 */
public class EditLogTailer {

    private final Logger logger = LoggerFactory.getLogger(EditLogTailer.class);

    private Configuration configuration = Configuration.getInstance();

    private long TAILEDITS_PERIOD = Long.parseLong(configuration.get("tailedits.period"));

    private long seq = 0L;

    private NameNodeRpcClient namenode;

    private FSNameSystem fsNameSystem;

    public EditLogTailer(NameNodeRpcClient namenode, FSNameSystem fsNameSystem) {
        this.fsNameSystem = fsNameSystem;
        this.namenode = namenode;
    }

    public void start() {
        EditLogTailerThread tailerThread = new EditLogTailerThread();
        tailerThread.start();
    }

    /**
     * 拉取edit log 操作
     */
    private void doTailEdits() {
       // logger.info("向namenode 发去edit logs seq = {} ....", seq);
        FetchEditLogResponse response = namenode.fetchEditLogs(seq);
        long ack = response.getAck();
        if (ack > seq) {
           // logger.info("本次拉取到的edit log 条数是 = {}", ack - seq);
            processEditLogs(response.getEditLogs());
            seq = ack;
        } else {
            // logger.info("没有拉取到一条edit log");
        }
    }

    /**
     * 处理拉取到的editlog
     *
     * @param editLogsJSON
     */
    private void processEditLogs(String editLogsJSON) {
        JSONArray editLogs = JSONArray.parseArray(editLogsJSON);
        for (int i = 0; i < editLogs.size(); i++) {
            JSONObject editsLog = editLogs.getJSONObject(i);
            String op = editsLog.getString("op");
            if (EditLogOperation.MKDIR.equals(op)) {
                String path = editsLog.getString("path");
                try {

                    long txid = editsLog.getLong("txid");
                    fsNameSystem.mkdir(path);
                    fsNameSystem.setMaxTxid(txid);

                    if (i == editLogs.size() - 1) {
                        logger.info("同步到的 edit log maxTxId = {}", txid);
                    }
                } catch (Exception e) {
                    logger.error("error", e);
                }
            }
        }
    }

    /**
     * 定时去namenode 拉取edit log 线程
     */
    private class EditLogTailerThread extends Thread {

        @Override
        public void run() {
            while (true) {
                doTailEdits();

                try {
                    Thread.sleep(TAILEDITS_PERIOD);
                } catch (Exception e) {
                    logger.error("error", e);
                }
            }
        }
    }


}
